package cn.cumulomq;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.region.policy.PolicyEntry;
import org.apache.activemq.broker.region.policy.PolicyMap;
import org.apache.activemq.broker.region.policy.StorePendingQueueMessageStoragePolicy;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter;
import org.apache.activemq.usage.MemoryUsage;
import org.apache.activemq.usage.StoreUsage;
import org.apache.activemq.usage.SystemUsage;
import org.apache.activemq.usage.TempUsage;
import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Logger;

import cn.ait.umc.endpoint.ConnectListener;
import cn.ait.umc.endpoint.DefaultEndpoint;
import cn.ait.umc.endpoint.Endpoint;
import cn.ait.umc.endpoint.IMessagePostSendProcessor;
import cn.ait.umc.endpoint.IMessagePreSendProcessor;
import cn.ait.umc.jms.EndpointMessageListener;
import cn.ait.umc.jms.JmsMessageTool;
import cn.ait.umc.logicalchannel.LogicalChannelConfig;
import cn.ait.umc.logicalchannel.SerialLogicalChannelConfig;
import cn.ait.umc.logicalchannel.TCPLogicalChannelConfig;
import cn.ait.umc.message.Message;
import cn.cumulomq.entity.Address;
import cn.cumulomq.entity.ChannelConfig;
import cn.cumulomq.exception.StartQMXException;
import cn.cumulomq.parse.ParseManage;
import cn.cumulomq.service.AddressService;
import cn.cumulomq.service.AftnChannelService;
import cn.cumulomq.service.ReceiveMessageService;

/**
 * 主程序入口，用于启动连接，注册监听
 * @author maxuedong
 */
public class MQXMain {
	Logger logger = Logger.getLogger(MQXMain.class);
	//系统只可以启动一次
	private boolean isStart=false;
	public Map<String,Endpoint> endpoints = new HashMap<String,Endpoint>();//所有的AFTN信道，KEY为信道ID
	public Map<String,Endpoint> addrEndpointMap = new HashMap<String,Endpoint>();//所有的AFTN信道，KEY为addr
	private Destination destination;
	private static MQXMain mqxMain = new MQXMain();
	private MQXMain(){
	}
	public static MQXMain getInstance(){
		return mqxMain;
	}
	/**
	 * 启动系统，建立信道连接，注册收报监听
	 * @throws StartQMXException
	 */
	public void start() throws StartQMXException{
		if(isStart)return;
		ParseManage.getInstance().init();
		startBroker();
		initReceiveMessageDestination();
		initReceiveMessageListener();
		startEndpoints();
		isStart=true;
	}
	
	private void startBroker(){
		new Thread(new Runnable() {
			@Override
			public void run() {
				logger.info("Start ActiveMQ begin...");
				BrokerService broker = new BrokerService();
				// configure the broker
				try {
					broker.setBrokerName("mqxbroker");
					broker.setUseJmx(true);
					broker.setPersistent(true);
					SystemUsage sysUsage = new SystemUsage();         
		            MemoryUsage memUsage = new MemoryUsage();  
		            StoreUsage storeUsage = new StoreUsage();  
		            TempUsage tmpUsage = new TempUsage();  
		              
		            memUsage.setLimit((1024 * 1024 * 1024)*2);
		            storeUsage.setLimit((1024 * 1024 * 1024) * 20);  
		            tmpUsage.setLimit((1024 * 1024 * 1024) * 10); 

		            sysUsage.setMemoryUsage(memUsage);  
		            sysUsage.setStoreUsage(storeUsage);  
		            sysUsage.setTempUsage(tmpUsage);
		            broker.setSystemUsage(sysUsage);
		            
		            PolicyMap policyMap = new PolicyMap();  
		            PolicyEntry pe = new PolicyEntry();  
		            pe.setOptimizedDispatch(true);  
		            pe.setProducerFlowControl(false);  
		            pe.setPrioritizedMessages(true);  
		            pe.setQueuePrefetch(1);  
		            pe.setUseCache(false);  
		            pe.setProducerFlowControl(true);  
		            pe.setMemoryLimit(1*1024*1024);  
		            pe.setCursorMemoryHighWaterMark(85);  
		            pe.setPendingQueuePolicy(new StorePendingQueueMessageStoragePolicy());  
		            policyMap.put(new ActiveMQQueue(">"), pe); 
		            broker.setDestinationPolicy(policyMap);
		            
					KahaDBPersistenceAdapter kahaPA = new KahaDBPersistenceAdapter();
					broker.setPersistenceAdapter(kahaPA);

					broker.addConnector("vm://localhost:61666");
					broker.start();
					logger.info("Start ActiveMQ end.");
					while(true){
						Thread.sleep(1000*1000);
					}
					
				} catch (Exception e) {
					logger.error("System start acitvemq error.",e);
				}
			}
		}).start();
		
	}
	private void initReceiveMessageListener(){
		ActiveMQConnectionFactory connectionFactory;
        // Connection ：JMS 客户端到JMS Provider 的连接
        Connection connection = null;
        // Session： 一个发送或接收消息的线程
        Session session;
        // Destination ：消息的目的地;消息发送给谁.
        // 消费者，消息接收者
        MessageConsumer consumer;
        connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD, "vm://localhost:61666");
        JmsMessageTool.getInstance().setJmsConnectionFactory(connectionFactory);
        int cnt=0;
        while(true){
        	cnt++;
            try {
    			connection = connectionFactory.createConnection();
    			session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
    			destination = session.createQueue("mqx_receivemsg");
    			consumer = session.createConsumer(destination);
    			consumer.setMessageListener(new ReceiveMessageService());
    			connection.start();
    			break;
    		} catch (JMSException e) {
    			try {
					Thread.sleep(1000);
				} catch (InterruptedException e1) {
				}
    		}
            if(cnt>30){
            	break;
            }
        }
	}
	private void initReceiveMessageDestination(){
		// ConnectionFactory ：连接工厂，JMS 用它创建连接
		ActiveMQConnectionFactory connectionFactory;
        // Connection ：JMS 客户端到JMS Provider 的连接
        Connection connection = null;
        // Session： 一个发送或接收消息的线程
        Session session;
        // Destination ：消息的目的地;消息发送给谁.
        
        connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD, "vm://localhost:61666");
//        JmsMessageTool.getInstance().setJmsConnectionFactory(connectionFactory);
        int cnt=0;
        while(true){
        	cnt++;
            try {
    			connection = connectionFactory.createConnection();
    			session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
    			destination = session.createQueue("mqx_receivemsg");
    			break;
    		} catch (JMSException e) {
    			try {
					Thread.sleep(1000);
				} catch (InterruptedException e1) {
				}
    		}
            if(cnt>30){
            	break;
            }
        }
	}
	/**
	 * 查询地址，启动信道
	 * @throws StartQMXException
	 */
	private void startEndpoints() throws StartQMXException{
		AddressService addrService = new AddressService();
		AftnChannelService chService = new AftnChannelService();
		List<Address> addrList = addrService.list(new Address());
		StringBuffer startErrorAddrs = new StringBuffer();

		for(ChannelConfig cc:chService.list(new ChannelConfig())){
			try {
				connectDMHS(cc);
			} catch (Throwable e) {
				startErrorAddrs.append(cc.getName()).append(":").append(cc.getId()).append(";");
				logger.error(cc.getName()+" connect error.",e);
			}			
		}
		
		for(Address addr:addrList){
			if(addr.getType().equals(Address.TYPE_AMHS)){//AMHS
				loginMTA(addr);
			}else{//AFTN or SITA
				if(StringUtils.isNotBlank(addr.getAftnChannelId())){
					try {
//						connectDMHS(chService.get(addr.getAftnChannelId()));
						addrEndpointMap.put(addr.getAddr(), endpoints.get(addr.getAftnChannelId()));
					} catch (Throwable e) {
						startErrorAddrs.append(addr.getName()).append(":").append(addr.getAddr()).append(";");
						logger.error(e);
					}
				}else{
					logger.warn("Address channel is null, address:"+addr.getAddr()+",id:"+addr.getId());
				}
			}
		}
		if(startErrorAddrs.toString().length()>0){
			throw new StartQMXException(startErrorAddrs.append(" start error.").toString());
		}
	}
	/**
	 * 连接MTA
	 * @param address
	 */
	void loginMTA(Address address){
		address.getAddr();
		address.getMtaIp();
		address.getAmhsUser();
		address.getAmhsPassword();
	}
	/**
	 * 连接转报机
	 * @param channelConfig
	 * @throws Throwable
	 */
	public synchronized void connectDMHS(final ChannelConfig channelConfig) throws Throwable{
	    LogicalChannelConfig lcConfig;
		if(ChannelConfig.COMMTYPE_AMTCP==channelConfig.getCommType()){
			lcConfig = new LogicalChannelConfig();
			lcConfig.setPhyInterface(channelConfig.getAmTcpInterface());
		}else if(ChannelConfig.COMMTYPE_SERIAL==channelConfig.getCommType()){
			lcConfig = new SerialLogicalChannelConfig();
			lcConfig.setPhyInterface(channelConfig.getSerialInterface());
		}else if(ChannelConfig.COMMTYPE_TCP==channelConfig.getCommType()){
			lcConfig = new TCPLogicalChannelConfig();
			lcConfig.setPhyInterface(channelConfig.getTcpInterface());
		}else{
			logger.warn("Channel config error,unknow commtype:"+channelConfig.getCommType()+",channel name is:"+channelConfig.getName());
			return;
		}
		lcConfig.setAftnProtocolParam(channelConfig.getMessageProtocolParameter());
		if(endpoints.get(channelConfig.getId())==null) {
			final DefaultEndpoint endpoint = new DefaultEndpoint();
			endpoints.put(channelConfig.getId(), endpoint);
			endpoint.setConnectListener(new ConnectListener());
			EndpointMessageListener dml = new EndpointMessageListener();
			dml.setDestination(destination);
			endpoint.setEndpointMessageListener(dml);
			endpoint.setMessagePostSendProcessor(new IMessagePostSendProcessor() {
				@Override
				public void process(Message arg0) {
					// TODO Auto-generated method stub
					
				}
			});
			endpoint.setMessagePreSendProcessor(new IMessagePreSendProcessor() {
				@Override
				public void process(Message arg0) {
					// TODO Auto-generated method stub
					
				}
			});
			
		    endpoint.setLcConfig(lcConfig);
//		    MessageUtil.replaceHeadAndEnd(endpoint);//替换报头报尾
		    new Thread(new Runnable() {
				public void run() {					
					try {
						endpoint.start();
					} catch (Throwable e) {
						logger.error(e.getMessage(),e);
					}
				}
			}).start();
		}else{
			
			if(!endpoints.get(channelConfig.getId()).isConnected()){
				endpoints.get(channelConfig.getId()).setLcConfig(lcConfig);
				new Thread(new Runnable() {
					public void run() {					
						try {
							endpoints.get(channelConfig.getId()).start();
							
						} catch (Throwable e) {
							logger.error(e.getMessage(),e);
						}
					}
				}).start();
				
			}
		}

	}
	public synchronized void disConnectDMHS(ChannelConfig channelConfig) throws Throwable{
		if(endpoints.get(channelConfig.getId())!=null) {
			endpoints.get(channelConfig.getId()).stop();
		}
	}
	public static void main(String args[]) throws IOException{
	/*	try {
			MQXMain.getInstance().start();
		} catch (StartQMXException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}*/
	}
}
